package cn.ipanel.bigdata.job.basic

import cn.ipanel.bigdata.boot.Job
import cn.ipanel.bigdata.boot.config.{ConfigKey, Configuration}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.dw.dim.phoenix.T_CLASS_TAG
import cn.ipanel.bigdata.utils.Dictionary
import cn.ipanel.bigdata.utils.Dictionary.F_DATE
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import org.apache.spark.sql.functions
import org.apache.spark.sql.functions.{col, lit, when}
import scalaj.http.Http

import scala.collection.mutable.ListBuffer


/**
 * class 跟 tag 的映射关系
 */
class Class2Tag extends Job {

  val T = T_CLASS_TAG

  override def onPrepare(): Unit = {
    T.deleteByFullDay(_DAY)
  }

  override def onStartup(): Unit = {
    val body = Http(Configuration.getParam(ConfigKey.URL_GET_DEVICE_TYPE))
      .charset("UTF-8")
      .asString.body

    val json = JSON.parseObject(body)
    if (json.getInteger("error_code") != null && json.getInteger("error_code") == 0) {
      val array: JSONArray = json.getJSONArray("result")
      val buf = ListBuffer[(String, String, String, String)]()
      for (idx <- 0 until array.size()) {
        val o = array.getJSONObject(idx)
        buf.append((o.getString("class"), o.getString("tag"), o.getString("title"), o.getString("tag_name")))
      }

      T.save(
        spark.createDataFrame(buf).toDF(T.F_CLASS, T.F_TAG, T.F_TITLE, T.F_TAG_NAME)
          .withColumn(F_DATE, lit(_DAY.toDate))
          // 可能有 tag 不全的问题，补充为 unknown
          .withColumn(T.F_TAG, when(col(T.F_TAG).isNull or col(T.F_TAG) === lit(""), lit(Dictionary.UNKNOWN)).otherwise(col(T.F_TAG)))
          // 将 class 和 tag 两遍的空格去掉
          .withColumn(T.F_TAG, functions.trim(col(T.F_TAG)))
          .withColumn(T.F_CLASS, functions.trim(col(T.F_CLASS)))
          .withColumn(T.F_TAG_NAME, functions.trim(col(T.F_TAG_NAME)))
      )
    } else {
      Logger.I("error code is not zero, err msg: " + json.getString("reply"))
    }
  }


}
